package com.craiditx.udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.xssf.usermodel.XSSFRow;
import org.apache.poi.xssf.usermodel.XSSFSheet;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;

import java.io.InputStream;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class trans_tag_mapping extends GenericUDTF {
    /**
     * 读取excel中标签数据
     */
    public List<Map<String,String>> getExcel() throws Exception{
        InputStream inputStream=this.getClass().getResourceAsStream("/trade_mapping_setting.xlsx");
        XSSFWorkbook sheets = new XSSFWorkbook(inputStream);
        XSSFSheet sheetAt = sheets.getSheetAt(0);
        List<Map<String,String>> excel_data=new ArrayList<>();
        for(int i=1;i<sheetAt.getPhysicalNumberOfRows();i++){
            XSSFRow row = sheetAt.getRow(i);
            Iterator<Cell> cellIterator = row.cellIterator();
            Map<String,String> cell_map=new HashMap<>();
            int count=1;
            while(cellIterator.hasNext()){
                Cell next = cellIterator.next();
                if(next.toString()==""){cell_map.put("lev"+count,"null");}
                else{cell_map.put("lev"+count,next.toString());}
                count++;
            }
            excel_data.add(cell_map);
        }
        inputStream.close();
        return excel_data;
    }

    /**
     * 剔除mapping映射数据中多余的 ｜ 符号
     */
    public static String clearRegex(String regex){
        char c=regex.charAt(0);
        char end=regex.charAt(regex.length()-1);
        StringBuffer buffer=new StringBuffer();
        buffer.append(c);
        for(int i=1;i<regex.length()-1;i++){
            if(c!='|' || regex.charAt(i)!='|'){buffer.append(regex.charAt(i));c=regex.charAt(i);}
        }
        if(end != '|'){buffer.append(end);}
        return buffer.toString();
    }

    /**
     *标签映射，入参为直接传入字符串,正则为excel中的商户关键字与经营范围
     */
    public static List<Map<String, String>> mappingFeature(String str) throws Exception {
        List<Map<String, String>> excel = new trans_tag_mapping().getExcel();
        List<Map<String,String>> mappingResult=new ArrayList<>();
        for (Map<String, String> map : excel) {
            String KEYWORD=map.get("lev4");
            String BUSINESS=map.get("lev5");
            Map<String,String> mapR=new HashMap<>();
            if(contData(str, clearRegex(BUSINESS))){
                mapR.put("text",str);
                mapR.put("lev1",map.get("lev1"));
                mapR.put("lev2",map.get("lev2"));
                mapR.put("lev3",map.get("lev3"));
                mappingResult.add(mapR);
            }else if(contData(str, clearRegex(KEYWORD))){
                mapR.put("text",str);
                mapR.put("lev1",map.get("lev1"));
                mapR.put("lev2",map.get("lev2"));
                mapR.put("lev3",map.get("lev3"));
                mappingResult.add(mapR);
            }
        }
        return mappingResult;
    }

    /**
     * 将集合转化为自定义拼接符字符串
     */
    public static String mkString(List<String> words, String mk){
        StringBuffer buffer=new StringBuffer();
        for (int i=0;i<words.size();i++) {
            if(i==words.size()-1) buffer.append(words.get(i));
            else buffer.append(words.get(i)+mk);
        }
        return buffer.toString();
    }

    /**
     * 数据清洗，正则清洗掉无用字符
     */
    public static String clearData(String data,String regex){
        String resultStr=data;
        Matcher matcher = Pattern.compile(regex).matcher(data);
        while(matcher.find()){
            String group = matcher.group();
            resultStr=resultStr.replace(group,"");
        }
        return resultStr;
    }

    /**
     *正则匹配字符串
     */
    public static boolean contData(String data,String regex){
        Matcher matcher = Pattern.compile(regex).matcher(data);
        boolean flag=false;
        while(matcher.find()){
            flag=true;
            break;
        }
        return flag;
    }

//    /**
//     * 通过配置文件获取配置信息
//     */
//    public Properties readProperty() throws Exception{
//        Properties proper=new Properties();
//        InputStream inputStream = this.getClass().getResourceAsStream("/application.properties");
//        InputStreamReader inputStreamReader=new InputStreamReader(inputStream,"UTF-8");
//        proper.load(inputStreamReader);
//        inputStream.close();
//        return proper;
//    }

    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        ArrayList<String> fieldNames = new ArrayList<String>();
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldNames.add("lev1");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldNames.add("lev2");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldNames.add("lev3");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    @Override
    public void process(Object[] args) throws HiveException {
        String input = args[0].toString();
        try {
            String etledStr = clearData(input,"[^\\u4e00-\\u9fa5\\d\\w\\-·[（][）][(][)]]|_");
            List<Map<String, String>> mf_word_list = mappingFeature(etledStr);
            // 行数是list的长度
            int rowNum = mf_word_list.size();
            for (int i = 0; i < rowNum; i++) {
                String lev1 = mf_word_list.get(i).get("lev1");
                String lev2 = mf_word_list.get(i).get("lev2");
                String lev3 = mf_word_list.get(i).get("lev3");
                String[] resultSet = new String[3];
                resultSet[0] = lev1;
                resultSet[1] = lev2;
                resultSet[2] = lev3;
//                System.out.println("一二三级标签是："+resultSet[0]+"\t"+resultSet[1]+"\t"+resultSet[2]);
                forward(resultSet);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws HiveException {

    }

    public static void main(String[] args) throws HiveException {
        trans_tag_mapping trans_tag_mapping = new trans_tag_mapping();
        String[] arr = {"中国人寿"};
        trans_tag_mapping.process(arr);
    }
}
